package com.ruyuan.event.streaming.process;
import com.ruyuan.event.streaming.pojo.EventJoinLog;
import com.ruyuan.event.streaming.pojo.EventServerLog;
import com.ruyuan.event.streaming.pojo.ProcessInfo;
import com.ruyuan.event.streaming.utils.Constants;
import com.ruyuan.event.streaming.utils.HBaseUtils;
import com.ruyuan.event.streaming.utils.KafkaProducerUtils;
import com.ruyuan.event.streaming.utils.RedisUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.kafka.clients.producer.Producer;
import redis.clients.jedis.Jedis;
import java.util.Iterator;

public class EventRetryJoinWindowFunction extends RichWindowFunction<EventJoinLog,EventJoinLog,String, TimeWindow> {
    String hbaseTableName = Constants.HBASE_EVENT_SERVER_LOG_TABLE;
    HTable hTable;
    Jedis jedis;
    Producer producer;

    @Override
    public void open(Configuration parameters) throws Exception {
        hTable = HBaseUtils.initHbaseClient(hbaseTableName);
        jedis = RedisUtils.getJedis();
        producer = KafkaProducerUtils.getProducer();
        super.open(parameters);
    }

    @Override
    public void apply(String s, TimeWindow window, Iterable<EventJoinLog> input, Collector<EventJoinLog> out) {
        Iterator<EventJoinLog> iterator = input.iterator();
        while (iterator.hasNext()) {
            EventJoinLog eventJoinLog = iterator.next();
            //重试的间隔时间必须要大于2秒
            if (System.currentTimeMillis() - eventJoinLog.getProcessInfo().getProcessTimestamp() > 2000) {
                byte[] key = ProcessETLUtils.generateBytesKey(eventJoinLog);
                //先重redis查询
                EventServerLog serverContext = ProcessETLUtils.getContext(jedis, key);
                //重置ProcessInfo
                EventJoinLog.Builder eventJoinBuilder = eventJoinLog.toBuilder();
                ProcessInfo.Builder processBuilder = eventJoinBuilder.getProcessInfo().toBuilder();
                processBuilder.setRetryCount(processBuilder.getRetryCount()+1);
                processBuilder.setProcessTimestamp(System.currentTimeMillis());
                if (serverContext == null) {
                    processBuilder.setJoinServerLog(false);
                }else {
                    processBuilder.setJoinServerLog(true);
                }
                eventJoinBuilder.setProcessInfo(processBuilder);

                //继续重试
                if (serverContext == null) {
                    //重试次数如果5次直接写入下游
                    EventJoinLog eventJoin = eventJoinBuilder.build();
                    if (eventJoin.getProcessInfo().getRetryCount() > 5) {
                        out.collect(eventJoinLog);
                    }else {
                        //否则的话继续重试
                        ProcessETLUtils.retry(producer,eventJoinLog.toByteArray());
                    }
                }else {
                    //重试成功，完成数据拼接
                    ProcessETLUtils.joinServerLogContext(eventJoinBuilder,serverContext);
                    out.collect(eventJoinLog);
                }
            }
        }
    }
}